package cn.edu.flink.scala.tutorial.source

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.scala._

object FromSourceTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val source = KafkaSource
      .builder[String]
      .setBootstrapServers("localhost:9092")
      .setTopics("test")
      .setGroupId("test")
      .setStartingOffsets(OffsetsInitializer.earliest())
      .build()

    val kafakaDS = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")

  }
}
